package com.alipay.service.job;


import com.alipay.constants.AuthenticationConstants;
import com.alipay.exception.MqProduceException;
import com.alipay.mapper.TransferTaskMapper;
import com.alipay.model.dto.TransDto;
import com.alipay.model.entity.TransTaskEntity;
import com.alipay.model.enums.QueueEnum;
import com.alipay.model.enums.TaskStatusEnum;
import com.alipay.util.DateUtil;
import com.alipay.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;

/**
 * 对A系统中非最终状态的任务消息表进行处理
 * @author Jun
 */
@Slf4j
@Service
public class TransferTaskDealJob {

    @Resource
    private TransferTaskMapper transferTaskMapper;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Scheduled(cron = "0 */1 * * * ?")
    public void processJob() {
        // 每1分钟执行一次，获取1分钟之前未达到最终态的任务, 具体时间间隔可以根据业务具体调整
        TransTaskEntity query = new TransTaskEntity();
        query.setStatus(TaskStatusEnum.SUCCESS.getCode());
        query.setCreateTime(DateUtil.getMinutesAgo(-1));

        // 每次处理一部分，不需要全部加载进内存
        List<TransTaskEntity> transTaskEntities = transferTaskMapper.getUnfinishedTransactionTask(query, AuthenticationConstants.LIMIT);
        if (CollectionUtils.isEmpty(transTaskEntities)) {
            return;
        }
        while (CollectionUtils.isNotEmpty(transTaskEntities)) {
            transTaskEntities.parallelStream().forEach(transTaskEntity -> {
                try {
                    String transId = transTaskEntity.getTransId();
                    if (StringUtils.isNotBlank(transId)) {
                        try {
                            TransDto transDto = new TransDto();
                            transDto.setAmount(transTaskEntity.getAmount());
                            transDto.setTransId(transTaskEntity.getTransId());
                            transDto.setFromAccount(transTaskEntity.getAccountId());
                            String toAccount = transTaskEntity.getRemark().split("-->")[1];
                            transDto.setToAccount(toAccount);
                            transDto.setAmount(transDto.getAmount());
                            // 对A系统中非最终状态的任务消息表进行处理, 这里需要用MQ的ack确认机制，保证消息发送成功
                            rabbitTemplate.convertAndSend(
                                    QueueEnum.INCREASE_QUEUE.getName(),
                                    JsonUtil.toJson(transDto));
                        } catch (Exception e) {
                            log.error("TransferTaskDealJob processJob failed, transId is {}, exception is ", transTaskEntity.getTransId(), e);
                            throw new MqProduceException("send message error");
                        }
                    }
                } catch (Exception e) {
                    log.error("TransferTaskDealJob processJob failed , transId is {}", transTaskEntity.getTransId(), e);
                }
            });

            // 再取下一部分数据
            transTaskEntities = transferTaskMapper.getUnfinishedTransactionTaskByLastId(query, transTaskEntities.get(transTaskEntities.size() - 1).getId(), AuthenticationConstants.LIMIT);
        }
    }


}
